package cn.xinfei.xdecision.data.thread;


import cn.xinfei.xdecision.common.model.datax.enums.ExecutorSceneEnum;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.*;


@Slf4j
public class AsyncExecuteThreadPool {

    private static final Logger LOGGER = LoggerFactory.getLogger(AsyncExecuteThreadPool.class);
    public static Map<String, TraceThreadPoolExecutor> threadPoolExecutorMap = new ConcurrentHashMap<>();
    private final TraceThreadPoolExecutor executorService;
    private final GracefulShutdownThread gracefulShutdownThread;
    private final BasicThreadFactory threadFactory;

    /**
     * 单位 : 秒
     */
    private static final int SHUTDOWN_TIMEOUT = 60;

    public AsyncExecuteThreadPool(
            int corePoolSize,
            int maximumPoolSize,
            int keepAliveTime,
            TimeUnit unit,
            BlockingQueue<Runnable> workQueue,
            String namePattern,
            RejectedExecutionHandler handler,
            Boolean independentTrace,
            ExecutorSceneEnum executorEnum) {
        this.executorService = new TraceThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                unit,
                workQueue,
                (new BasicThreadFactory.Builder()).namingPattern(namePattern).build(),
                handler,
                independentTrace,
                executorEnum,
                namePattern.replaceAll("%d", ""));
        threadPoolExecutorMap.put(namePattern.replaceAll("%d", ""), this.executorService);
        this.threadFactory = (BasicThreadFactory) executorService.getThreadFactory();
        this.gracefulShutdownThread = new GracefulShutdownThread();
        Runtime.getRuntime().addShutdownHook(this.gracefulShutdownThread);
    }

    class GracefulShutdownThread extends Thread {

        @Override
        public void run() {
            shutdown(SHUTDOWN_TIMEOUT);
        }
    }

    public void shutdown(long timeout) {
        //shutdown executor
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(timeout, TimeUnit.SECONDS)) {
                log.debug("executor has not terminated.forcing shutdown of executor");
                executorService.shutdownNow();
                while (!executorService.awaitTermination(timeout, TimeUnit.SECONDS)) {
                    log.debug("waiting for executor to terminate");
                }
            }
            if (executorService.awaitTermination(timeout, TimeUnit.SECONDS)) {
                log.info("AsyncExecutePool shutdown cleanly for {}", threadFactory.getNamingPattern().replaceAll("-%d", ""));
            } else {
                LOGGER.error("Timed out waitting for AsyncExecutePool {} to shutdown ,exit uncleanly", threadFactory.getNamingPattern().replaceAll("-%d", ""));
            }
        } catch (InterruptedException e) {
            log.error("interrupted while waiting for executor to terminate");
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            LOGGER.error("Error when AsyncExecutePool {} for shutting down", threadFactory.getNamingPattern().replaceAll("-%d", ""));
        }
    }

    public void execute(Runnable command) {
        this.executorService.execute(command);
    }

    public void shutdown() {
        this.executorService.shutdown();
    }

    public Future submit(Runnable command) {
        return this.executorService.submit(command);
    }

    public <T> Future<T> submit(Callable<T> command) {
        return this.executorService.submit(command);
    }

    public ThreadPoolExecutor getExecutorService() {
        return this.executorService;
    }

}
